package day9;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class Test3_SQL {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        DataStreamSource<String> ds1 = env.socketTextStream("hadoop10", 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> ds2 = ds1.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                String[] arr = value.split(",");
                return new Tuple2(arr[0], Integer.parseInt(arr[1]));
            }
        });

        // 1、SQL语法 ：将dataStream 转成 table1
        /*tenv.createTemporaryView("table1",ds2, $("word"),$("cont"));

        Table table2 = tenv.sqlQuery("select word,sum(cnt) c1\n" +
                "from table1\n" +
                "group by word");*/

        // 2、table API
        Table table = tenv.fromDataStream(ds2, $("word"),$("cnt"));
        Table table2 = table.groupBy($("word")).select($("word"),$("cnt").sum().as("cnt2"));

        // 把上述查询结果转成DataStream再进行打印
        DataStream<Tuple2<Boolean, Tuple2<String, Integer>>> ds3 =
                tenv.toRetractStream(table2, TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}));

        ds3.print();
    }
}
